package com.yudaxian.bi.bizmq;

import com.rabbitmq.client.Channel;
import com.yudaxian.bi.common.Const;
import com.yudaxian.bi.common.ErrorCode;
import com.yudaxian.bi.exception.BusinessException;
import com.yudaxian.bi.model.entity.Chart;
import com.yudaxian.bi.model.enums.ChartStatusEnum;
import com.yudaxian.bi.model.query.ChartQuery;
import com.yudaxian.bi.service.ChartService;
import com.yudaxian.bi.utils.SparkAiUtils;
import com.yudaxian.bi.utils.StringTools;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @Author: yudaxian
 * @Description: TODO
 * @DateTime: 2024/9/22 23:32
 **/
@Component
@Slf4j
public class BiMessageConsumer {

    @Resource
    private ChartService chartService;

    @Resource
    private SparkAiUtils sparkAiUtils;

    @SneakyThrows
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = BiMqConstant.BI_QUEUE_NAME, durable = "true"),
            exchange = @Exchange(value = BiMqConstant.BI_EXCHANGE_NAME),
            key = {BiMqConstant.BI_ROUTING_KEY}
    ))
    public void receiveMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        log.info("Received message: {}, deliveryTag: {}", message, deliveryTag);
        if (StringTools.isEmpty(message)) {
            channel.basicNack(deliveryTag, false, false);
            throw new BusinessException(ErrorCode.SYSTEM_ERROR, "消息为空");
        }
        Long chartId = Long.valueOf(message);

        Chart chart = chartService.getChartById(Long.valueOf(chartId));

        if (chart == null) {
            channel.basicNack(deliveryTag, false, false);
            throw new BusinessException(ErrorCode.PARAMS_ERROR, "消息为空");
        }

        // 先修改图表任务状态为“执行中”，执行成功后，修改为“已完成”，执行失败后，修改为“失败”，记录失败信息
        Chart updateChart = new Chart();
        updateChart.setStatus(ChartStatusEnum.PROCESSING.getCode());
        log.info("{}",chart.getId());
        ChartQuery updateChartQuery = new ChartQuery();
        updateChartQuery.setStatus(ChartStatusEnum.WAITING.getCode());
        updateChartQuery.setId(chart.getId());
        Integer updateByParamCount = chartService.updateByParam(updateChart, updateChartQuery);
        if (updateByParamCount == 0) {
            channel.basicNack(deliveryTag, false, false);
            handleError(updateChartQuery.getId(), "更新图表状态{等待}失败");
            return;
        }
        String result = sparkAiUtils.sendMessage(userInput(chart));
        // 获取option
//		int option = result.indexOf("option");
//		String substring = result.substring(option, result.length());
        String[] split = result.split("=====");
        if (split.length != 3) {
            String answer = "根据这个回答模板 { " + Const.ANSWER_MODEL +" }, 帮我分析一下这个数据 ," + userInput(chart);
            String resultMessage = sparkAiUtils.sendMessage(answer);
            split = resultMessage.split("=====");
        }
        if (split.length != 3) {
            channel.basicNack(deliveryTag, false, false);
            handleError(updateChartQuery.getId(), "AI智能分析失败，格式有误");
            return;
        }
        if (StringTools.isEmpty(split[1].trim()) || StringTools.isEmpty(split[2].trim())) {
            channel.basicNack(deliveryTag, false, false);
            handleError(updateChartQuery.getId(), "AI智能分析失败，数据为空");
            return;
        }
        updateChart.setGenChart(split[1].trim());
        updateChart.setGenResult(split[2].trim());
        updateChart.setStatus(ChartStatusEnum.FINISHED.getCode());
        updateChartQuery.setStatus(ChartStatusEnum.PROCESSING.getCode());
        updateByParamCount = chartService.updateByParam(updateChart, updateChartQuery);
        if (updateByParamCount == 0) {
            channel.basicNack(deliveryTag, false, false);
            handleError(updateChartQuery.getId(), "更新图表状态{成功}失败");
        }

        log.info("Received message: {}, deliveryTag: {}", chartId, deliveryTag);
        channel.basicAck(deliveryTag, false);
    }

    private String userInput(Chart chart) {
        String goal = chart.getGoal();
        String chartType = chart.getChartType();
        String csv = chart.getChartData();
        StringBuilder sb = new StringBuilder();
        sb.append(Const.GEN_GOAL).append("\n");
        if (!StringTools.isEmpty(chartType)) {
            goal += Const.CHART_PREFIX + chartType;
        }
        sb.append(goal).append("\n");
        sb.append(Const.GEN_DATA).append("\n").append(csv).append("\n");
        return sb.toString();
    }

    private void handleError(Long chartId, String errMessage) {
        Chart chart = new Chart();
        chart.setStatus(ChartStatusEnum.ERROR.getCode());
        chart.setMessage(errMessage);
        ChartQuery chartQuery = new ChartQuery();
        chartQuery.setStatus(ChartStatusEnum.PROCESSING.getCode());
        chartQuery.setId(chartId);
        Integer updateByParamCount = chartService.updateByParam(chart, chartQuery);
        if (updateByParamCount == 0) {
            log.error("更新{失败}状态失败，chartId={}, errMessage={}", chartId, errMessage);
        }
    }

}
